package com.shujia.flink.sink;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;

public class Demo02MySink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        ArrayList<String> arrList = new ArrayList<>();
        arrList.add("flink");
        arrList.add("flink");
        arrList.add("flink");
        arrList.add("flink");
        DataStreamSource<String> ds = env.fromCollection(arrList);

        ds.addSink(new MySinkFunction());

        env.execute();


    }
}

class MySinkFunction implements SinkFunction<String>{

    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.println("进入了invoke方法");
        // invoke 每一条数据会执行一次
        // 最终数据需要sink到哪里，就对value进行处理即可
        System.out.println(value);
    }
}